package com.ada.spark.rddoperator

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 作用：类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]。
  * 假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。
  */
object Spark03_mapPartitions {

    def main(args: Array[String]): Unit = {
        //创建SparkConf
        val conf = new SparkConf().setAppName("Spark03_mapPartitions").setMaster("local[*]")
        //创建Spark上下文对象
        val sc = new SparkContext(conf)

        val listRDD: RDD[Int] = sc.makeRDD(1 to 5)

        //mapPartitions可以对一个RDD中所有的分区进行遍历
        //mapPartitions效率优于map算子，减少了发送到执行器执行交互次数
        //mapPartitions可能会出现内存溢出（OOM）
        val mapPartitionsRDD: RDD[Int] = listRDD.mapPartitions(datas => {
            datas.map(data => data * 2)
        })

        println(mapPartitionsRDD.collect().mkString(","))
        //2,4,6,8,10

        /**
          * map()和mapPartition()的区别
          * 1、map()：每次处理一条数据。
          * 2、mapPartition()：每次处理一个分区的数据，这个分区的数据处理完后，原RDD中分区的数据才能释放，可能导致OOM。
          * 3、开发指导：当内存空间较大的时候建议使用mapPartition()，以提高处理效率。
          */
    }
}
